package com.common.boot.common;

import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

public class BachUtil {
  /** 批量保护数量 */
  public static final Integer OPEN_BACH_PROTECT = 996;
  /** 批次数量 */
  public static final Integer NUMBER_BACH_PROTECT = 996;
  /** 多线程方式 注意： 1. 线程数量，根据自己的环境，自行调节。这里是根据系统核心数给的默认值。 2. 连接池数量，线程池数量过大，容易撑爆线程池，所以适当调节线程池大小。 */
  public static final ExecutorService THREAD_POOL = Executors.newWorkStealingPool();

  /**
   * 批量中包含有关联查询，可能因为数据过长出现数据查询的的问题， 这里进行分批处理
   *
   * @param list 需要处理的list
   * @param bach 批量处理的函数
   * @param <T> 元素类型
   */
  public static <T> void protectBach(List<T> list, Consumer<List<T>> bach) {
    if (isEmpty(list)) {
      return;
    }
    if (list.size() > OPEN_BACH_PROTECT) {
      for (int i = 0; i < list.size(); i += NUMBER_BACH_PROTECT) {
        int lastIndex = Math.min(i + NUMBER_BACH_PROTECT, list.size());
        bach.accept(list.subList(i, lastIndex));
      }
    } else {
      bach.accept(list);
    }
  }

  /**
   * 批量中包含有关联查询，可能因为数据过长出现数据查询的的问题 这里进行分批，合并数据处理。
   *
   * @param list 原数据
   * @param bach 处理方法
   * @param <T> 原数据类型
   * @param <R> 处理结果类型
   * @return 处理结果汇总
   */
  public static <T, R> List<R> bach(List<T> list, Function<List<T>, List<R>> bach) {
    if (isEmpty(list)) {
      return Collections.emptyList();
    }
    if (list.size() > OPEN_BACH_PROTECT) {
      List<R> end = new LinkedList<>();
      for (int i = 0; i < list.size(); i += NUMBER_BACH_PROTECT) {
        int lastIndex = Math.min(i + NUMBER_BACH_PROTECT, list.size());
        Optional.ofNullable(bach.apply(list.subList(i, lastIndex))).ifPresent(end::addAll);
      }
      return end;
    }
    return bach.apply(list);
  }

  private static <T> boolean isEmpty(Collection<T> collection) {
    return collection == null || collection.isEmpty();
  }

  public static <T, R> List<R> bachByThread(List<T> list, Function<List<T>, List<R>> bach) {
    if (isEmpty(list)) {
      return Collections.emptyList();
    }
    if (list.size() > OPEN_BACH_PROTECT) {
      List<Future<List<R>>> futures = new LinkedList<>();
      for (int i = 0; i < list.size(); i += NUMBER_BACH_PROTECT) {
        int finalI = i;
        int lastIndex = Math.min(i + NUMBER_BACH_PROTECT, list.size());
        futures.add(THREAD_POOL.submit(() -> bach.apply(list.subList(finalI, lastIndex))));
      }
      return futures.parallelStream()
          .flatMap(
              a -> {
                try {
                  return Optional.ofNullable(a.get()).orElse(Collections.emptyList()).stream();
                } catch (InterruptedException | ExecutionException e) {
                  throw new RuntimeException(e);
                }
              })
          .collect(Collectors.toList());
    }
    return bach.apply(list);
  }
}
